package ds_industry_2025.ds.YangJuan_2024.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

/*
    5、根据dws层表来计算每个地区2020年订单金额前3省份，依次存入MySQL数据库shtd_result的regiontopthree表中（表结构如下），
    然后在Linux的MySQL命令行中根据地区表主键升序排序，查询出前5条，将SQL语句复制粘贴至客户端桌面【Release\任务B提交结果.docx】
    中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下；
字段	类型	中文含义	备注
regionid	int	地区表主键
regionname	text	地区名称
provinceids	text	省份表主键	用,分割显示前三省份的id
provincenames	text	省份名称	用,分割显示前三省份的name
provinceamount	text	省份名称	用,分割显示前三省份的订单金额（需要去除小数部分，使用四舍五入）

例如：
3	华东地区	21,27,11	上海市,江苏省,浙江省	100000,100,10
 */
object t17 {
  def main(args: Array[String]): Unit = {
   val spark=SparkSession.builder()
      .master("local[*]")
      .appName("t2")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

   spark.table("dwd.fact_order_info")
     .where("etl_date=(select max(etl_date) from dwd.fact_order_info)")
     .where(year(col("create_time"))=== 2020)
     .createOrReplaceTempView("o")

    spark.table("dwd.dim_province")
      .where("etl_date=(select max(etl_date) from dwd.fact_order_info)")
      .createOrReplaceTempView("p")

    spark.table("dwd.dim_region")
      .where("etl_date=(select max(etl_date) from dwd.dim_region)")
      .createOrReplaceTempView("r")

    val r1 = spark.sql(
      """
        |select distinct
        |region_id,region_name,province_id,province_name,
        |row_number() over(partition by region_id,region_name order by money desc) as row,
        |money
        |from(
        |select distinct
        |p.region_id,
        |r.region_name,
        |o.province_id,
        |p.name as province_name,
        |round(sum(o.final_total_amount) over(partition by p.region_id,r.region_name,o.province_id,p.name),0) as money
        |from o
        |join p on p.id=o.province_id
        |join r on r.id=p.region_id
        |) as r1
        |""".stripMargin)

    val result = r1.filter(col("row") < 4)
      .groupBy("region_id", "region_name")
      .agg(
        concat_ws(",", collect_list("province_name")).as("province_names"),
        concat_ws(",", collect_list("province_id")).as("province_ids"),
        concat_ws(",", collect_list("money")).as("moneys")
      )
      .select("region_id","region_name","province_ids","province_names","moneys")


    result.show(false)


    spark.close()
  }

}
